package org.tong.kafka.consumers;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

@Slf4j
//@Component
public class MyConsumer {

    @KafkaListener(topics = {"test"}, groupId = "batch-g1")
    public void consumerMsgBatch(List<String> records, Acknowledgment ack) {
        log.info("批量消费消息开始");
        log.info(records.toString());
        log.info("批量消费消息结束");
        ack.acknowledge();
    }

    @KafkaListener(topics = {"test-topic"}, groupId = "test-g1")
    public void consumerMsgSingle(List<String> records, Acknowledgment ack) {
        log.info("单条消费消息开始");
        log.info(records.toString());
        log.info("单条消费消息结束");
        ack.acknowledge();
    }
}
